package cn.wangjie.spark.etl

import cn.itcast.spark.utils.IpUtils
import cn.wangjie.spark.config.ApplicationConfig
import cn.wangjie.spark.utils.{IpUtils, SparkUtils}
import org.apache.spark.SparkFiles
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{StringType, StructType}
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
import org.lionsoul.ip2region.{DbConfig, DbSearcher}

/**
 * 广告数据进行ETL处理，具体步骤如下：
 * 第一步、加载json数据
	 * 第二步、解析IP地址为省份和城市
	 * 第三步、数据保存至Hive表
 */
object PmtEtlRunner {
	
	/**
	 * 对数据进行ETL处理，调用第三方ip2Region库解析IP地址为省份和城市
	 */
	def processData(dataframe: DataFrame): DataFrame = {
		// 获取SparkSession对象
		val session: SparkSession = dataframe.sparkSession
		import session.implicits._
		
		// TODO: 当对DataFrame使用类似RDD中转换函数时，建议先转换为RDD，再调用函数，安全
		// step1: 对RDD进行转换
		val newRDD: RDD[Row] = dataframe.rdd.mapPartitions{ iter =>
			// 创建DbSearcher对象，每个分区创建一个
			val dbSearcher: DbSearcher = new DbSearcher(new DbConfig(), SparkFiles.get("ip2region.db"))
			
			// 对每个分区中数据，解析IP值
			iter.map{row =>
				// 获取IP值
				val ipValue: String = row.getAs[String]("ip")
				// 调用工具类解析
				val region: Region = IpUtils.convertIpToRegion(ipValue, dbSearcher)
				// 将解析获取的省份和城市，添加到原来每行数据Row中
				val newSeq = row.toSeq :+ region.province :+ region.city
				// 返回新的Row
				Row.fromSeq(newSeq)
			}
		}
		
		// step2： 自定义Schema
		val newSchema: StructType = dataframe.schema
    		.add("province", StringType, nullable = true)
    		.add("city", StringType, nullable = true)
		
		// step 3: 结合RDD[Row]和Schema创建DataFrame
		val df: DataFrame = session.createDataFrame(newRDD, newSchema)
		
		// TODO: step4: 分区表的数据，添加分区列，表示数据所属日期并返回
		df.withColumn("date_str", date_sub(current_date(), 1).cast(StringType))
	}
	
	/**
	 * 将DataFrame数据保存至Parquet列式存储文件中
	 */
	def saveAsParquet(dataframe: DataFrame): Unit = {
		dataframe
			// 降低分数数目
    		.coalesce(1)
    		.write
    		.mode(SaveMode.Overwrite)
			// 指定分区列
    		.partitionBy("date_str")
    		.parquet("dataset/pmt-etl")
	}
	
	/**
	 * 保存数据至Hive分区表中，按照日期字段分区
	 */
	def saveAsHiveTable(dataframe: DataFrame): Unit = {
		dataframe
			// 降低分数数目
			.coalesce(1)
			.write
			.mode(SaveMode.Append)
			// 指定分区列
			.partitionBy("date_str")
			// TODO: 必须指定format数据源为hive，否则有异常
    		.format("hive")
    		.saveAsTable("itcast_ads.pmt_ads_info")
	}
	
	
	def main(args: Array[String]): Unit = {
		
		// 设置Spark应用程序运行的用户：root, 默认情况下为当前系统用户
		System.setProperty("user.name", "root")
		System.setProperty("HADOOP_USER_NAME", "root")
		
		// 1. 创建SparkSession实例对象
		val spark: SparkSession = SparkUtils.createSparkSession(this.getClass)
		import spark.implicits._
		
		// TODO：将IP地址解析字典数据文件进行分布式缓存
		spark.sparkContext.addFile(ApplicationConfig.IPS_DATA_REGION_PATH)
		
		// 2. 加载json数据
		val pmtDF: DataFrame = spark.read.json(ApplicationConfig.DATAS_PATH)
		//pmtDF.printSchema()
		//pmtDF.show(10, truncate = false)
		
		// 3. 解析IP地址为省份和城市
		val etlDF: DataFrame = processData(pmtDF)
		//etlDF.printSchema()
		//etlDF.show(10, truncate = false)
		
		// 4. 保存ETL数据至Hive分区表
		//saveAsParquet(etlDF)
		saveAsHiveTable(etlDF)
		
		// 5. 应用结束，关闭资源
		//Thread.sleep(100000)
		spark.stop()
	}
	
}
